package com.fury.demos.workers;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class WorkerDemo {

	private static final int NUMBER_OF_JOBS = 1000 * 1000;
	private static final int WORKERS = Runtime.getRuntime().availableProcessors();

	private final CyclicBarrier barrier;
	private final CyclicBarrier goBarrier;
	private final BlockingQueue<Job> queue;
	private final CyclicBarrier reinitBarrier;

	/*
	 * Start the application.
	 */
	public static void main(String[] args) throws InterruptedException {
		new WorkerDemo();
	}

	/*
	 * The main class that sets up the barriers, queue, buffer, workers and runs
	 * the main logic.
	 */
	public WorkerDemo() throws InterruptedException {
		// We'll use this to synchronize start of all threads (including this
		// one).
		goBarrier = new CyclicBarrier(WORKERS + 1);

		// We'll use this to synchronize the end of all threads execution cycle.
		barrier = new CyclicBarrier(WORKERS + 1);

		// We'll use this to synchronize the reset of the other barriers.
		reinitBarrier = new CyclicBarrier(WORKERS + 1);

		// All workers drain from this queue, and execute the jobs.
		queue = new ArrayBlockingQueue<Job>(NUMBER_OF_JOBS);

		// We wish to populate this buffer with a number of jobs, and then add
		// it all in one atomic call to the queue.
		Collection<Job> buffer = new ArrayList<Job>(NUMBER_OF_JOBS);
		for (int i = 0; NUMBER_OF_JOBS > i; i++) {
			buffer.add(new Job());
		}

		// For comparison, let's see how fast it is on a single thread.
		long start = System.nanoTime();
		for (Job job : buffer) {
			job.execute();
		}
		System.out.printf("1 thread takes %f ms.\n", (float) (System.nanoTime() - start) / 1000000f);

		// Create and start the threads. Note, they will not start executing
		// until they get unblocked by the goBarrier.
		Thread[] workThreads = new Thread[WORKERS];
		for (int i = 0; WORKERS > i; i++) {
			workThreads[i] = new Thread(new Worker());
			workThreads[i].start();
		}

		while (true) {
			System.out.println("");
			System.out.println(System.currentTimeMillis() + " New run");
			// Add buffer to queue in a single atomic action.
			start = System.nanoTime();
			queue.addAll(buffer);
			System.out.printf(System.currentTimeMillis() + " Adding to queue took %f ms.\n", (float) (System.nanoTime() - start) / 1000000f);

			// This thread gives the go ahead to all other threads that are
			// ready
			// (all must be in order for all to run).
			// We wish to block all threads until the queue has finished the
			// addAll,
			// that is added ALL the items and not just the first few.
			try {
				goBarrier.await();
			} catch (BrokenBarrierException e1) {
				e1.printStackTrace();
			}

			System.out.println(System.currentTimeMillis() + " Started");

			// ... worker threads executing the jobs somewhere here ...

			// Now we wait until all jobs have been executed, the barrier will
			// unblock when all worker threads are done.
			try {
				barrier.await();
			} catch (BrokenBarrierException e) {
				e.printStackTrace();
			}

			// We should reach here when all jobs are done!
			System.out.println(System.currentTimeMillis() + " All done");

			goBarrier.reset();
			barrier.reset();

			System.out.println(System.currentTimeMillis() + " Done resetting");

			try {
				reinitBarrier.await();
			} catch (BrokenBarrierException e) {
				e.printStackTrace();
			}

			System.out.println(System.currentTimeMillis() + " Done waiting for reinitBarrier");

			reinitBarrier.reset();

			System.out.println(System.currentTimeMillis() + " end");

		}

	}

	/*
	 * A dummy Job class that loops a number of times to simulate CPU load.
	 */
	private class Job {
		public void execute() {
			int a = 0;
			for (int i = 0; 1000 > i; i++) {
				a += i;
			}
		}
	}

	/*
	 * Worker class that drains from the queue and runs the jobs, while being in
	 * sync with other threads and primary thread.
	 */
	private class Worker implements Runnable {
		private static final int TODRAIN = 100;
		private final List<Job> drainer;

		public Worker() {
			this.drainer = new ArrayList<Job>(TODRAIN);
		}

		@Override
		public void run() {
			while (true) {
				long start = System.nanoTime();
				int jobCounter = 0;
				int drainCounter = 0;

				// We wish to wait until all workers are ready, but primarily
				// wait for the final go-ahead when the buffer has been added to
				// the queue.
				try {
					goBarrier.await();
				} catch (InterruptedException e1) {
					e1.printStackTrace();
				} catch (BrokenBarrierException e1) {
					e1.printStackTrace();
				}

				// Drain from the queue a number of jobs, until the queue is
				// empty.
				drainer: while (true) {
					if (queue.drainTo(drainer, TODRAIN) > 0) {
						drainCounter++;
						for (Job job : drainer) {
							job.execute();
							jobCounter++;
						}
						drainer.clear();
					} else {
						break drainer;
					}
				}

				// There are no more drains to be done, the queue is empty.
				if (drainer.isEmpty()) {
					float total = (float) (System.nanoTime() - start) / 1000000f;
					System.out.printf(System.currentTimeMillis() + " " + Thread.currentThread().getName() + " executed " + jobCounter + " jobs in " + drainCounter
							+ " drains and in total of %f ms.\n", total);

					try {
						barrier.await();
					} catch (InterruptedException e) {
						e.printStackTrace();
					} catch (BrokenBarrierException e) {
						e.printStackTrace();
					}
				}

				try {
					reinitBarrier.await();
				} catch (InterruptedException e) {
					e.printStackTrace();
				} catch (BrokenBarrierException e) {
					e.printStackTrace();
				}
			}

		}

	}

}
